package com.legsem.jhosync.util;

import java.util.Arrays;
import java.util.Map;
import java.util.Map.Entry;
import java.util.TreeMap;

import com.legsem.jhosync.change.ChangesSet;
import com.legsem.jhosync.change.Deletes;
import com.legsem.jhosync.change.Inserts;

/**
 * Data returned by sources is stored as a document in a special format that
 * <ul>
 * <li>helps with finding differences between 2 data sets</li>
 * <li>reduce the size of the serialized data</li>
 * <ul/>
 * For this to work, sources do not return just any type of structure. They
 * should return key/attribute/value tuples.
 * <p/>
 * This class provides methods to encode/decode key/attribute/values to hashmaps
 * and also compare 2 sets of key/attribute/values.
 */
public class KAVUtil {

    /** Key/attribute/value separator character. */
    public static final String KAV_SEP = ":";

    /** Character that separates one key/attribute/value from the next one. */
    public static final String KAVS_SEP = ">#<";

    /** Key/attribute/value string format. */
    public static final String KAV_FORMAT = KAVS_SEP + "%s" + KAV_SEP + "%s"
            + KAV_SEP + "%s";

    /**
     * Tries to simulate the behavior of redis#sdiff.
     * <p/>
     * Assuming 2 sets of character strings organized as a series of KAVS_SEP
     * separated character strings, return the changes, in the form of inserts
     * and deletes, that needs to be applied to the second set in order to make
     * it identical to the first one.
     * <p/>
     * * For example:
     * <ul>
     * <li>set 1: <code>string 1KAVS_SEPstring 2KAVS_SEPstring 3</code></li>
     * <li>set 2: <code>string 2KAVS_SEPstring 4</code></li>
     * </ul>
     * sdiff(set1, set2) returns <code>string 1KAVS_SEPstring 3</code> as
     * inserts (meaning they need to be inserted in set 2) and
     * <code>string 4</code> as deletes (meaning they need to be removed from
     * set 2.
     * 
     * @param set1 first key/attribute/value set
     * @param set2 second key/attribute/value set
     * @param inserts the result set of inserts to apply to set 2
     * @param deletes the result set of deletes to apply to set 2
     */
    public static void sdiff(String set1, String set2, Inserts inserts,
            Deletes deletes) {

        String[] a1 = (set1 == null) ? new String[0] : set1.split(KAVS_SEP);
        String[] a2 = (set2 == null) ? new String[0] : set2.split(KAVS_SEP);
        Arrays.sort(a1);
        Arrays.sort(a2);
        int i1 = 0;
        int i2 = 0;

        while (i1 < a1.length && i2 < a2.length) {
            int r = a1[i1].compareTo(a2[i2]);
            if (r == 0) {
                i1++;
                i2++;
            } else if (r < 0) {
                addKav(inserts.get(true), a1[i1]);
                i1++;
            } else {
                addKav(deletes.get(true), a2[i2]);
                i2++;
            }
        }

        while (i1 < a1.length) {
            addKav(inserts.get(true), a1[i1]);
            i1++;
        }

        while (i2 < a2.length) {
            addKav(deletes.get(true), a2[i2]);
            i2++;
        }

    }

    /**
     * Add an encoded key/attribute/value to a hash of hashes.
     * <p/>
     * Ignores non encoded strings
     * <p/>
     * The value might contain the separator, in which case we reconstruct it.
     * 
     * @param map the hash of hashes being populated
     * @param kavEncoded the encoded key/attribute/value tuple
     */
    public static void addKav(Map < String, Map < String, Object >> map,
            String kavEncoded) {
        String[] kav = kavEncoded.split(KAV_SEP);
        if (kav.length == 3) {
            addKav(map, kav[0], kav[1], kav[2]);
        } else if (kav.length > 3) {
            StringBuilder sb = new StringBuilder(kav[2]);
            for (int i = 3; i < kav.length; i++) {
                sb.append(KAV_SEP);
                sb.append(kav[i]);
            }
            addKav(map, kav[0], kav[1], sb.toString());
        }
    }

    /**
     * Add a key/attribute/value tuple to a hash of hashes.
     * 
     * @param map the hash of hashes being populated
     * @param key the tuple key
     * @param attribute the tuple attribute
     * @param value the tuple value
     */
    public static void addKav(Map < String, Map < String, Object >> map,
            String key, String attribute, Object value) {
        Map < String, Object > attrs = map.get(key);
        if (attrs == null) {
            attrs = new TreeMap < String, Object >();
            map.put(key, attrs);
        }
        attrs.put(attribute, value);
    }

    /**
     * Remove a key/attribute/value tuple from a map.
     * 
     * @param map the hash of hashes being updated
     * @param key the tuple key
     * @param attribute the tuple attribute to remove
     */
    public static void delKav(Map < String, Map < String, Object >> map,
            String key, String attribute) {
        Map < String, Object > attrs = map.get(key);
        if (attrs != null) {
            attrs.remove(attribute);
        }

    }

    /**
     * Items from the input map are encoded as key/attribute/value tuples
     * separated by line feeds (\n).
     * <p/>
     * We start by sorting keys in alphabetical order in order to guarantee that
     * the serialized character string is always the same whatever the original
     * hashmap sort order was.
     * <p/>
     * FIXME what happens if value contains semicolon?
     * 
     * @param map a map of maps where primary keys are associated with maps
     *            holding attribute/value pairs (column/value pairs).
     * @return a key/attribute/value encoding of the input map
     */
    public static String encode(Map < String, Map < String, Object > > map) {
        if (map == null) {
            return null;
        }
        StringBuilder sb = new StringBuilder();
        Map < String, Map < String, Object >> sortedMap = new TreeMap < String, Map < String, Object >>(
                map);
        for (Entry < String, Map < String, Object > > record : sortedMap
                .entrySet()) {
            Map < String, Object > sortedAttrs = new TreeMap < String, Object >(
                    record.getValue());
            for (Entry < String, Object > attribute : sortedAttrs.entrySet()) {
                sb.append(encode(record.getKey(), attribute.getKey(),
                        attribute.getValue()));
            }
        }
        return (sb.length() == 0) ? "" : sb.substring(KAVS_SEP.length());
    }

    /**
     * Encode a key/attribute/value tuple.
     * 
     * @param key of the record holding attributes
     * @param attributeName the attribute name
     * @param attributeValue the attribute value
     * @return an encoded key/attribute/value
     */
    public static String encode(String key, String attributeName,
            Object attributeValue) {
        return String.format(KAV_FORMAT, key, attributeName,
                attributeValue.toString());
    }

    /**
     * Decode data encoded using kavEncode back to a map of maps.
     * 
     * FIXME what happens if values contain line feeds?
     * 
     * @param encoded the encoded data
     * @return a map of maps where primary keys are associated with maps holding
     *         attribute/value pairs (column/value pairs).
     */
    public static Map < String, Map < String, Object > > decode(String encoded) {
        if (encoded == null) {
            return null;
        }
        Map < String, Map < String, Object > > map = new TreeMap < String, Map < String, Object > >();
        if (encoded.length() > 0) {
            String[] kavs = encoded.split(KAVS_SEP);
            for (String kav : kavs) {
                addKav(map, kav);
            }
        }
        return map;
    }

    /**
     * Apply changes to a KAV encoded set.
     * 
     * @param changes the insert and delete operations to apply
     * @param set the key/attribute/value encoded set
     * @return the updated set
     */
    public static String apply(ChangesSet changes, String set) {
        Map < String, Map < String, Object > > map = null;
        if (set == null) {
            map = new TreeMap < String, Map < String, Object > >();
        } else {
            map = decode(set);
        }
        Map < String, Map < String, Object > > inserts = changes.getInserts()
                .get();
        Map < String, Map < String, Object > > deletes = changes.getDeletes()
                .get();

        if (deletes != null) {
            for (Entry < String, Map < String, Object > > rec : deletes
                    .entrySet()) {
                for (Entry < String, Object > attr : rec.getValue().entrySet()) {
                    delKav(map, rec.getKey(), attr.getKey());
                }
            }
        }
        if (inserts != null) {
            for (Entry < String, Map < String, Object > > rec : inserts
                    .entrySet()) {
                for (Entry < String, Object > attr : rec.getValue().entrySet()) {
                    addKav(map, rec.getKey(), attr.getKey(), attr.getValue());
                }
            }
        }
        return encode(map);

    }
}
